package com.sunday.common.mq.kafka.study.spring.e3_Sending_Messages;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.AlternativeJdkIdGenerator;
import org.springframework.util.IdGenerator;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.concurrent.*;

/**
 * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
 */
@Slf4j
@RestController
@RequestMapping("/test")
//public class MQApi implements SmartInitializingSingleton {
public class MQApi {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * ====================================== KafkaTemplate ================================
     */
    @GetMapping("/param")
    public void param() {
        log.info("DefaultTopic : {}", kafkaTemplate.getDefaultTopic());
    }

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#kafka-template
     */
    @GetMapping("/sendDefault")
    public void sendDefault() throws ExecutionException, InterruptedException {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.sendDefault("sendDefault(V data)");
        log.info("{}", future.get());
        future = kafkaTemplate.sendDefault("key", "sendDefault(K key, V data)");
        log.info("{}", future.get());
        future = kafkaTemplate.sendDefault(2, "key", "sendDefault(Integer partition, K key, V data)");
        log.info("{}", future.get());
        future = kafkaTemplate.sendDefault(2, System.currentTimeMillis(), "key", "sendDefault(Integer partition, Long timestamp, K key, V data)");
        log.info("{}", future.get());
    }

    @GetMapping("/send/{key}")
    public void send(@PathVariable String key) throws ExecutionException, InterruptedException {
        CompletableFuture<SendResult<String, String>> future;
        future = kafkaTemplate.send("topic1", key);
        log.info("{}", future.get());
        future = kafkaTemplate.send("topic1", key, "send(String topic, K key, V data)");
        log.info("{}", future.get());
        future = kafkaTemplate.send("topic1", 3, "key", "send(String topic, Integer partition, K key, V data)");
        log.info("{}", future.get());
        future = kafkaTemplate.send("topic1", 2, System.currentTimeMillis(), "key", "send(String topic, Integer partition, Long timestamp, K key, V data)");
        log.info("{}", future.get());
        future = kafkaTemplate.send(new ProducerRecord<>("topic1", key, "ProducerRecord"));
        log.info("{}", future.get());

        /**
         * 方式 1
         */
        IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
        Message<String> message1 = MessageBuilder.withPayload("GenericMessage")
                .setHeader(KafkaHeaders.TOPIC, "topic1")
                .setHeader(KafkaHeaders.PARTITION, 4)
                .setHeader(KafkaHeaders.KEY, key)
                .setHeader(KafkaHeaders.CORRELATION_ID, defaultIdGenerator.generateId())
                .build();
        future = kafkaTemplate.send(message1);
        log.info("{}", future.get());
        /**
         * 方式 2
         */
        Message<String> message2 = MessageBuilder.createMessage("GenericMessage", new MessageHeaders(Map.of(
                KafkaHeaders.TOPIC, "topic1",
                KafkaHeaders.PARTITION, 4,
                KafkaHeaders.KEY, key
        )));
        future = kafkaTemplate.send(message2);
        log.info("{}", future.get());
        /**
         * 方式 3
         */
        Message<String> message3 = new GenericMessage<>("GenericMessage", Map.of(
                KafkaHeaders.TOPIC, "topic1",
                KafkaHeaders.PARTITION, 4,
                KafkaHeaders.KEY, key
        ));
        future = kafkaTemplate.send(message3);
        log.info("{}", future.get());
    }

    @GetMapping("/metrics")
    public void metrics() {
        kafkaTemplate.metrics().forEach((metricName, metric) -> {
            log.info("{}", metricName);
            log.info("{}", metric);
        });
    }

    @GetMapping("/partitionsFor/{topic}")
    public void partitionsFor(@PathVariable String topic) {
        kafkaTemplate.partitionsFor(topic).stream().forEach(partitionInfo -> {
            log.info("{}", partitionInfo.toString());
        });
    }

    /**
     * 引发问题，spring kafka是如何保证消息在没有达到batch size准实时发送得
     * <p>
     * 资料 1
     * 只有当
     * {@link ProducerFactory}提供一个单例的生产者(比如
     * {@link DefaultKafkaProducerFactory})。
     * <p>
     * 资料 2
     * kafkaTemplate.flush() 是一个方法调用，用于强制将所有待发送的 Kafka 消息立即发送到 Kafka 代理。当使用 Spring Kafka 客户端进行消息生产时，通常会将消息写入内部缓冲区，然后定期或在特定条件下批量发送这些消息。但是，如果你想要立即发送所有待发送的消息，可以使用 flush() 方法。
     * 该方法将阻塞当前线程直到所有消息都成功发送到 Kafka 代理或发生错误。通过调用 flush()，你可以确保消息及时传递给 Kafka，而不必等待缓冲区中的消息自动发送。
     * <p>
     * 资料 3
     * https://kafka.apache.org/documentation/#producerconfigs_linger.ms
     * linger.ms
     * 生成者将两次请求传输之间到达的任何记录组合到单个批处理请求中。
     * 通常，只有当记录到达速度快于发送速度时，才会在负载下发生这种情况。
     * 但是，在某些情况下，即使在中等负载下，客户端也可能希望减少请求数。
     * 此设置通过添加少量人为延迟来实现此目的，也就是说，生成者将等待给定的延迟，以允许发送其他记录，以便可以将发送一起批处理，
     * 而不是立即发送记录。这可以被认为是类似于 TCP 中 Nagle 的算法。此设置给出了批处理延迟的上限：
     * 一旦我们获得了分区的记录值，无论此设置如何，它都会立即发送，但是，如果我们为该分区累积的字节数少于此数量，
     * 我们将“徘徊”指定时间等待更多记录出现。此设置默认为 0（即无延迟）。例如，设置 ，将减少发送的请求数，
     * 但会给在没有负载的情况下发送的记录增加多达 5 毫秒的延迟。batch.size.linger.ms=5
     * <p>
     * 在Spring Kafka中，可以通过以下方式来保证消息在没有达到batch size时的准实时发送：
     * linger.ms参数：通过设置linger.ms参数，可以控制消息在缓冲区等待的时间。当消息到达时，如果当前缓冲区中的消息数量未达到batch size，消息会被暂存，并等待一段时间以便更多消息到达。这样可以在一定程度上提高消息发送的效率和吞吐量。
     * max.request.size参数：通过适当调整max.request.size参数，可以增加每个请求中允许的最大字节数。这样即使单个消息的大小超过了batch size，仍然可以将其作为一个独立的请求发送出去，而不必等待其他消息。
     * 异步发送和回调函数：Spring Kafka的生产者API支持异步发送消息，并提供回调函数来处理发送结果。使用异步发送可以避免阻塞应用程序，并允许继续发送其他消息，即使当前批次大小未达到。
     * 发送策略和优化：根据具体场景，可以根据业务需求和性能要求进行发送策略和优化。例如，可以根据不同的消息类型或关键性等级划分不同的主题或分区，从而灵活地控制消息的发送行为。
     * 需要注意的是，准实时发送是在系统和网络条件允许的情况下实现的，并受到Kafka本身性能和配置参数的影响。因此，对于更高的实时性要求，可以考虑使用较小的batch size、较短的linger.ms时间和适当调整其他相关参数来提高消息的实时性。
     */
    @GetMapping("/flush")
    public void flush() {
        kafkaTemplate.flush();
    }

    /**
     * org.springframework.kafka.core.KafkaTemplate#doSend(org.apache.kafka.clients.producer.ProducerRecord, io.micrometer.observation.Observation)
     */
    @GetMapping("/execute")
    public void execute() throws ExecutionException, InterruptedException {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.execute(producer -> {
            final CompletableFuture<SendResult<String, String>> completableFuture = new CompletableFuture<>();
            ProducerRecord producerRecord = new ProducerRecord<>("topic1", "ProducerRecord");
            /**
             * org.springframework.kafka.core.KafkaTemplate#buildCallback(org.apache.kafka.clients.producer.ProducerRecord, org.apache.kafka.clients.producer.Producer, java.util.concurrent.CompletableFuture, java.lang.Object, io.micrometer.observation.Observation)
             */
            Future<RecordMetadata> sendFuture = producer.send(producerRecord, (metadata, exception) -> {
                if (exception == null) {
                    completableFuture.complete(new SendResult<>(producerRecord, metadata));
                } else {
                    completableFuture.completeExceptionally(
                            new KafkaProducerException(producerRecord, "Failed to send", exception));
                }
            });
            if (sendFuture.isDone()) {
                try {
                    sendFuture.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new KafkaException("Interrupted", e);
                } catch (ExecutionException e) {
                    throw new KafkaException("Send failed", e.getCause()); // NOSONAR, stack trace
                }
            }
            return completableFuture;
        });
        log.info("{}", future.get());
    }

    /**
     * 重点 1
     * <p>
     * 模拟启动两个服务节点
     * 例如：
     * A 节点
     * myId: partitions assigned: [topic1-5, topic1-6, topic1-7, topic1-8, topic1-9]
     * myId1: partitions assigned: [topic1-5, topic1-6, topic1-7, topic1-8, topic1-9]
     * myId2: partitions assigned: [topic1-0, topic1-1, topic1-2, topic1-3, topic1-4]
     * B 节点
     * myId: partitions assigned: [topic1-0, topic1-1, topic1-2, topic1-3, topic1-4]
     * myId1: partitions assigned: [topic1-0, topic1-1, topic1-2, topic1-3, topic1-4]
     * myId2: partitions assigned: [topic1-5, topic1-6, topic1-7, topic1-8, topic1-9]
     * <p>
     * 模拟发送一个消息， 返回响应参数 其中 recordMetadata=topic1-8@0
     * SendResult [
     * producerRecord=ProducerRecord(
     * topic=topic1,
     * partition=null,
     * headers=RecordHeaders(headers = [], isReadOnly = true),
     * key=null, value=send(String topic, V data), timestamp=null),
     * recordMetadata=topic1-8@0]
     * <p>
     * A节点获取到
     * [0]======= send(String topic, V data)
     * [1]======= send(String topic, V data)
     * B 节点获取到
     * [2]======= send(String topic, V data)
     * 符合topic1-8@0的划分，并且相同值的时候，均进行同样分配规则
     * <p>
     * 重点 2
     * Kafka通过键（key）来划分消息到不同的分区（partition）。每个主题（topic）在Kafka中都可以被划分为一个或多个分区，而每个分区是一个有序、持久化且可扩展的日志。
     * Kafka使用一种称为"键哈希"的机制将具有相同键的消息发送到同一个分区。这确保了所有具有相同键的消息将按照它们的键值顺序存储在同一分区内，并保证了在同一分区内的消息顺序性。
     * 分区在Kafka集群中以分布式方式进行管理，每个分区可分布在不同的代理节点上。这样的设计允许Kafka实现高吞吐量和扩展性，因为消息的读写操作可以并行处理，而不会受到单个节点的限制。
     * 通过将消息分配到多个分区，Kafka还实现了水平扩展和负载均衡。消费者可以并行地从多个分区中读取数据，提高了整体系统的处理能力和效率。
     * <p>
     * 重点 3
     * 在使用Spring Kafka时，当您调用send(String topic, V data)方法发送消息并没有指定键（key）时，默认情况下，Kafka会根据一种轮询（round-robin）的方式将消息均匀地分配到可用的分区。
     * 轮询分区策略意味着每次发送消息时，Kafka会按照顺序选择下一个可用的分区来存储消息。这样可以确保消息在所有可用分区之间均匀分布，从而实现负载均衡。
     * 需要注意的是，由于轮询分区策略不依赖于消息内容，因此无法保证具有相同内容的消息始终被分配到同一个分区。如果需要确保具有相同键的消息被发送到同一个分区，建议在发送消息时显式地指定键值。
     * 如果您希望自定义分区策略或按特定规则将消息路由到分区，可以通过实现Partitioner接口来自定义分区器，并在生产者配置中进行设置。这样您可以根据消息的特征、业务逻辑等自定义确定分区的算法策略。
     * 其中 生产者实例未改变：如果您在测试中使用的是同一个生产者实例发送多条消息，那么都会落到一个分区中，可以启动多个生产者实例进行观察。
     */
    @KafkaListener(id = "myId", topics = "topic1")
    public void listen(String in) {
        log.info("[0]======= {}", in);
    }

    @KafkaListener(id = "myId1", topics = "topic1")
    public void listen1(String in) {
        log.info("[1]======= {}", in);
    }

    @KafkaListener(id = "myId2", topics = "topic1")
    public void listen2(String in) {
        log.info("[2]======= {}", in);
    }

    /**
     * 当.partitions(1)的情况下，启动两个消费者
     * A 节点
     * myId_2: partitions assigned: [topic2-0]
     * B 节点
     * myId_2: partitions assigned: []
     * 发现当超过分区数量的时候，多余的消费者将没有任何动作
     * 当 A 节点失效后， B 节点获得分区 myId_2: partitions assigned: [topic2-0]
     */
    @GetMapping("/send_2/{key}")
    public void send_2(@PathVariable String key) throws ExecutionException, InterruptedException {
        CompletableFuture<SendResult<String, String>> future;
        future = kafkaTemplate.send("topic2", key, "send_2");
        log.info("{}", future.get());
    }

    @KafkaListener(id = "myId_2", topics = "topic2")
    public void listen_2(String in) {
        log.info("[0]======= {}", in);
    }

    /**
     * 当.partitions(3)的情况下，启动两个消费者，存在奇数轮
     * A 节点
     * myId_3: partitions assigned: [topic3-0, topic3-1]
     * B 节点
     * myId_3: partitions assigned: [topic3-2]
     * 发现当超分区数量超过消费者 出现不平均现象
     */
    @GetMapping("/send_3/{key}")
    public void send_3(@PathVariable String key) throws ExecutionException, InterruptedException {
        CompletableFuture<SendResult<String, String>> future;
        future = kafkaTemplate.send("topic3", key, "send_3");
        log.info("{}", future.get());
    }

    @KafkaListener(id = "myId_3", topics = "topic3")
    public void listen_3(String in) {
        log.info("[0]======= {}", in);
    }

//    /**
//     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#kafka-template
//     * 您可以配置 with a 以获取包含发送结果（成功或失败）的异步回调，而不是等待 完成。 下面的清单显示了接口的定义：KafkaTemplateProducerListenerFutureProducerListener
//     */
//    @Override
//    public void afterSingletonsInstantiated() {
//        kafkaTemplate.setProducerListener(new ProducerListener<>() {
//            @Override
//            public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
//                log.info("[onSuccess] producerRecord : {}, recordMetadata : {}", producerRecord, recordMetadata);
//            }
//
//            @Override
//            public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
//                log.error("[onSuccess] producerRecord : {}, recordMetadata : {}", producerRecord, recordMetadata, exception);
//            }
//        });
//    }
//
//    @GetMapping("/send_async/{key}")
//    public void send_async(@PathVariable String key) {
//        kafkaTemplate.send("topic3", key, "send_3");
//        log.info("============结束访问================");
//    }

    /**
     * send 方法返回 . 您可以向监听器注册回调，以异步方式接收发送结果。
     * 以下示例演示如何执行此操作：CompletableFuture<SendResult>
     * <p>
     * future.whenComplete()方法接受两个参数：result表示异步操作的结果，throwable表示可能发生的异常。这个方法会在当前线程上执行回调函数。
     * future.whenCompleteAsync()方法也接受相同的参数，但它会在一个新的线程或线程池中异步执行回调函数。
     */
    @GetMapping("/send_whenComplete/{key}")
    public void send_whenComplete(@PathVariable String key) {
        CompletableFuture<SendResult<String, String>> future;
        future = kafkaTemplate.send("topic3", key, "send_3");
        log.info("============结束访问================");
        future.whenComplete((result, throwable) -> log.info("[whenComplete] result : {}, throwable : {}", result, throwable));
        future.whenCompleteAsync((result, throwable) -> log.info("[whenCompleteAsync]result : {}, throwable : {}", result, throwable));
    }

    /**
     * 非阻塞（异步)
     */
    @GetMapping("/async/{key}")
    public void async(@PathVariable String key) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send("topic3", key, "send_3");
        future.whenComplete((result, throwable) -> log.info("[whenComplete]result"));
    }

    /**
     * 阻止（同步）
     */
    @GetMapping("/sync/{key}")
    public void sync(@PathVariable String key) throws ExecutionException, InterruptedException, TimeoutException {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send("topic3", key, "send_3");
        future.get(10, TimeUnit.SECONDS);
    }


}
